package com.ferry.transport.netty.channel;

import com.ferry.common.util.SystemClock;
import com.ferry.common.util.concurrent.AtomicUpdater;
import com.ferry.common.util.sequence.IntSequence;
import com.ferry.transport.api.Directory;
import com.ferry.transport.api.UnResolvedAddress;
import com.ferry.transport.api.channel.WrappedChannel;
import com.ferry.transport.api.channel.WrappedChannelGroup;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.netty.channel.ChannelFutureListener;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by daiyong
 */
public class NettyChannelGroup implements WrappedChannelGroup {

	private static long LOSS_INTERVAL = TimeUnit.MINUTES.toMillis(5);

	private static final int CPUS = Runtime.getRuntime().availableProcessors();

	private static int DEFAULT_SEQUENCE_STEP = (CPUS << 3) + 1;

	private static AtomicReferenceFieldUpdater<CopyOnWriteArrayList, Object[]> channelsUpdater =
			AtomicUpdater.newAtomicReferenceFieldUpdater(CopyOnWriteArrayList.class, Object[].class, "array");

	private static final AtomicIntegerFieldUpdater<NettyChannelGroup> signalNeededUpdater =
			AtomicIntegerFieldUpdater.newUpdater(NettyChannelGroup.class, "signalNeeded");

	private volatile int signalNeeded = 0;

	private final ConcurrentLinkedQueue<Runnable> waitAvailableListeners = new ConcurrentLinkedQueue<>();

	private UnResolvedAddress address;

	private final CopyOnWriteArrayList<NettyChannel> channels = new CopyOnWriteArrayList<>();

	private final ChannelFutureListener remover = future -> remove(NettyChannel.attachChannel(future.channel()));

	private final IntSequence sequence = new IntSequence(DEFAULT_SEQUENCE_STEP);

	private final ConcurrentMap<String, Integer> weights = Maps.newConcurrentMap();

	private final ReentrantLock lock = new ReentrantLock();

	private final Condition notifyCondition = lock.newCondition();

	private volatile int capacity = Integer.MAX_VALUE;

	private volatile int warmUp = 10 * 60 * 1000;  //预热时间

	private volatile long timestamp = SystemClock.millisClock().time();

	private volatile long deadlineMillis = -1;

	private volatile boolean connecting = false;

	public NettyChannelGroup(UnResolvedAddress address) {
		this.address = address;
	}

	@Override
	public UnResolvedAddress remoteAddress() {
		return address;
	}

	@Override
	public WrappedChannel next() {
		for (;;) {
			Object[] elements = channelsUpdater.get(channels);
			int length = elements.length;
			if (length == 0) {
				if (waitForAvailable(1000)) { // wait a moment
					continue;
				}
				throw new IllegalStateException("No channel");
			}
			if (length == 1) {
				return (WrappedChannel) elements[0];
			}

			int index = sequence.next() & Integer.MAX_VALUE;

			return (WrappedChannel) elements[index % length];
		}
	}

	@Override
	public List<? extends WrappedChannel> channels() {
		return Lists.newArrayList(channels);
	}

	@Override
	public boolean isEmpty() {
		return channels.isEmpty();
	}

	@Override
	public boolean add(WrappedChannel channel) {
		boolean added = channel instanceof NettyChannel && channels.add((NettyChannel) channel);
		if (added) {
			timestamp = SystemClock.millisClock().time(); // reset timestamp

			((NettyChannel) channel).channel().closeFuture().addListener(remover);

			deadlineMillis = -1;

			if (signalNeededUpdater.getAndSet(this, 0) != 0) { // signal needed: true
				lock.lock();
				try {
					notifyCondition.signalAll(); // must signal all
				} finally {
					lock.unlock();
				}
			}

			notifyListeners();
		}
		return added;
	}

	@Override
	public boolean remove(WrappedChannel channel) {
		boolean removed = channel instanceof NettyChannel && channels.remove(channel);
		if (removed) {
			timestamp = SystemClock.millisClock().time(); // reset timestamp

			if (channels.isEmpty()) {
				deadlineMillis = SystemClock.millisClock().time() + LOSS_INTERVAL;
			}
		}
		return removed;
	}

	void notifyListeners() {
		for (;;) {
			Runnable listener = waitAvailableListeners.poll();
			if (listener == null) {
				break;
			}
			listener.run();
		}
	}

	@Override
	public int size() {
		return channels.size();
	}

	@Override
	public void setCapacity(int capacity) {
		this.capacity = capacity;
	}

	@Override
	public int getCapacity() {
		return capacity;
	}

	@Override
	public boolean isConnecting() {
		return connecting;
	}

	@Override
	public void setConnecting(boolean connecting) {
		this.connecting = connecting;
	}

	@Override
	public boolean isAvailable() {
		return !channels.isEmpty();
	}

	@Override
	public boolean waitForAvailable(long timeoutMillis) {
		boolean available = isAvailable();
		if (available) {
			return true;
		}
		long remains = TimeUnit.MILLISECONDS.toNanos(timeoutMillis);

		lock.lock();
		try {
			// avoid "spurious wakeup" occurs
			while (!(available = isAvailable())) {
				signalNeeded = 1; // set signal needed to true
				if ((remains = notifyCondition.awaitNanos(remains)) <= 0) {
					break;
				}
			}
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		} finally {
			lock.unlock();
		}

		return available;
	}

	@Override
	public void onAvailable(Runnable listener) {
		waitAvailableListeners.add(listener);
		if (isAvailable()) {
			notifyListeners();
		}
	}

	@Override
	public int getWeight(Directory directory) {
		Integer weight = weights.get(directory.directoryString());
		return weight == null ? 50 : weight;
	}

	@Override
	public void putWeight(Directory directory, int weight) {
		if (weight == 50) {
			return;
		}
		weights.put(directory.directoryString(), weight > 100 ? 100 : weight);
	}

	@Override
	public void removeWeight(Directory directory) {
		weights.remove(directory.directoryString());
	}

	@Override
	public int getWarmUp() {
		return warmUp > 0 ? warmUp : 0;
	}

	@Override
	public void setWarmUp(int warmUp) {
		this.warmUp = warmUp;
	}

	@Override
	public boolean isWarmUpComplete() {
		return SystemClock.millisClock().time() - timestamp > warmUp;
	}

	@Override
	public long timestamp() {
		return timestamp;
	}

	@Override
	public long deadlineMillis() {
		return deadlineMillis;
	}

	@Override
	public boolean equals(Object o) {
		if (this == o) return true;
		if (o == null || getClass() != o.getClass()) return false;

		NettyChannelGroup that = (NettyChannelGroup) o;

		return address.equals(that.address);
	}

	@Override
	public int hashCode() {
		return address.hashCode();
	}

	@Override
	public String toString() {
		return "NettyChannelGroup{" +
				"address=" + address +
				", channels=" + channels +
				", weights=" + weights +
				", warmUp=" + warmUp +
				", timestamp=" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSSZ").format(new Date(timestamp)) +
				", deadlineMillis=" + deadlineMillis +
				'}';
	}

}
